'=====================================================================
'
'   File:   RemoveDuplicates.cs
'   Summary:    This file contains the implementation of the RemoveDuplicates
'           data flow transformation component. This component has a single input
'           and two outputs. The outputs are asynchronous, and the component caches the rows
'           it receives in its input buffer in ProcessInput until it receives notification from 
'           The data flow task that no more rows are coming. Then it sorts the cached rows, and directs duplicate
'           rows to its DuplicateOutput, and distinct rows to its distinct output.
'   Date:  6/15/2004
'
'---------------------------------------------------------------------
'
'   This file is part of the Microsoft SQL Server Code Samples.
'   Copyright (C) Microsoft Corporation.  All rights reserved.
'
'   This source code is intended only as a supplement to Microsoft
'   Development Tools and/or on-line documentation.  See these other
'   materials for detailed information regarding Microsoft code samples.
'
'   THIS CODE AND INFORMATION ARE PROVIDED "AS IS" WITHOUT WARRANTY OF 
'   ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO 
'   /THE IMPLIED WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A
'   PARTICULAR PURPOSE.
'
'===================================================================== 

'/ <summary>
'/ This sample demonstrates a data flow transformation component with asynchronous outputs.
'/ </summary>
<DtsPipelineComponent(DisplayName:="RemoveDuplicatesVB", Description:="Finds and removes duplicate rows.", IconResource:="Microsoft.Samples.SqlServer.Dts.RemoveDuplicates.ico", ComponentType:=ComponentType.Transform)> _
Public Class RemoveDuplicates
    Inherits PipelineComponent
    Private fireEventAgain As Boolean
    Private cancelEvent As Boolean
    Private inputBufferCache As Buffer
    Private colInfoDistinctOutput As ColumnInfos
    Private colInfoDuplicateOutput As ColumnInfos
    Private distinctBuffer As PipelineBuffer
    Private duplicatesBuffer As PipelineBuffer

    Public Overloads Overrides Sub ProvideComponentProperties()
        MyBase.RemoveAllInputsOutputsAndCustomProperties()
        MyBase.ProvideComponentProperties()
        ComponentMetaData.InputCollection(0).Name = "Input"
        Dim distinctOutput As IDTSOutput100 = ComponentMetaData.OutputCollection(0)
        distinctOutput.Name = "DistinctOutput"
        distinctOutput.Description = "Distinct rows are directed to this output."
        distinctOutput.SynchronousInputID = 0
        Dim duplicateOutput As IDTSOutput100 = ComponentMetaData.OutputCollection.New
        duplicateOutput.Name = "DuplicatesOutput"
        duplicateOutput.Description = "Duplicate rows are directed to this output."
        AddIsDistinctCustomProperty(distinctOutput, True)
        AddIsDistinctCustomProperty(duplicateOutput, False)
    End Sub

    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId:="Microsoft.Samples.SqlServer.Dts.RemoveDuplicates.InternalFireError(System.String)")> _
    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")> _
    <CLSCompliant(False)> _
    Public Overloads Overrides Function Validate() As DTSValidationStatus
        If Not (ComponentMetaData.OutputCollection.Count = 2) Then
            InternalFireError("There should be exactly two outputs. The metadata of this component is corrupt.")
            Return DTSValidationStatus.VS_ISCORRUPT
        End If

        If Not (ComponentMetaData.InputCollection.Count = 1) Then
            InternalFireError("There is more than one input. The metadata of this component is corrupt.")
            Return DTSValidationStatus.VS_ISCORRUPT
        End If

        Dim input As IDTSInput100 = ComponentMetaData.InputCollection(0)
        Dim distinct As IDTSOutput100 = ComponentMetaData.OutputCollection(0)
        Dim duplicate As IDTSOutput100 = ComponentMetaData.OutputCollection(1)

        If Not (distinct.CustomPropertyCollection.Count = 0) Then
            Try
                Dim isDistinct As IDTSCustomProperty100 = distinct.CustomPropertyCollection("IsDistinctOutput")
                If CType(isDistinct.Value, Boolean) = False Then
                    InternalFireError("The IsDistinct property is set incorrectly.")
                    Return DTSValidationStatus.VS_NEEDSNEWMETADATA
                End If
            Catch
                InternalFireError("The IsDistinct property has been removed from the component.")
                Return DTSValidationStatus.VS_NEEDSNEWMETADATA
            End Try
        Else
            InternalFireError("The IsDistinct property has been removed from the component.")
            Return DTSValidationStatus.VS_NEEDSNEWMETADATA
        End If

        If Not (duplicate.CustomPropertyCollection.Count = 0) Then
            Try
                Dim isDistinct As IDTSCustomProperty100 = duplicate.CustomPropertyCollection("IsDistinctOutput")
                If CType(isDistinct.Value, Boolean) = True Then
                    InternalFireError("The IsDistinct property is set incorrectly.")
                    Return DTSValidationStatus.VS_NEEDSNEWMETADATA
                End If
            Catch
                InternalFireError("The IsDistinct property has been removed from the component.")
                Return DTSValidationStatus.VS_NEEDSNEWMETADATA
            End Try
        Else
            InternalFireError("The IsDistinct property has been removed from the component.")
            Return DTSValidationStatus.VS_NEEDSNEWMETADATA
        End If

        If Not (input.InputColumnCollection.Count = distinct.OutputColumnCollection.Count) Then
            ComponentMetaData.FireError(0, ComponentMetaData.Name, "The number of input columns do not match the number of output columns in the " + distinct.Name + " output.", "", 0, cancelEvent)
            Return DTSValidationStatus.VS_NEEDSNEWMETADATA
        End If

        If Not (input.InputColumnCollection.Count = duplicate.OutputColumnCollection.Count) Then
            ComponentMetaData.FireError(0, ComponentMetaData.Name, "The number of input columns do not match the number of output columns in the " + duplicate.Name + " output.", "", 0, cancelEvent)
            Return DTSValidationStatus.VS_NEEDSNEWMETADATA
        End If

        For Each col As IDTSInputColumn100 In input.InputColumnCollection
            If Not OutputColumnExists(distinct, col.LineageID) Then
                ComponentMetaData.FireError(0, ComponentMetaData.Name, "The input column " + col.Name + " is not found in the " + distinct.Name + " output.", "", 0, cancelEvent)
                Return DTSValidationStatus.VS_NEEDSNEWMETADATA
            End If

            If Not OutputColumnExists(duplicate, col.LineageID) Then
                ComponentMetaData.FireError(0, ComponentMetaData.Name, "The input column " + col.Name + " is not found in the " + duplicate.Name + " output.", "", 0, cancelEvent)
                Return DTSValidationStatus.VS_NEEDSNEWMETADATA
            End If
        Next

        Return MyBase.Validate
    End Function

    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes")> _
    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId:="System.Exception.#ctor(System.String)")> _
    <CLSCompliant(False)> _
    Public Overloads Overrides Function SetUsageType(ByVal inputID As Integer, ByVal virtualInput As IDTSVirtualInput100, ByVal lineageID As Integer, ByVal usageType As DTSUsageType) As IDTSInputColumn100
        If virtualInput Is Nothing Then
            Throw New ArgumentNullException("virtualInput")
        End If

        Dim vCol As IDTSVirtualInputColumn100 = virtualInput.VirtualInputColumnCollection.GetVirtualInputColumnByLineageID(lineageID)
        Dim col As IDTSInputColumn100 = Nothing

        If vCol.DataType = DataType.DT_IMAGE Then
            Throw New Exception("Binary data types not supported.")
        End If

        If usageType = DTSUsageType.UT_IGNORED Then
            col = MyBase.SetUsageType(inputID, virtualInput, lineageID, usageType)
            RemoveOutputColumn(ComponentMetaData.OutputCollection(0).ID, vCol.LineageID)
            RemoveOutputColumn(ComponentMetaData.OutputCollection(1).ID, vCol.LineageID)
        Else
            If usageType = DTSUsageType.UT_READWRITE Then
                Throw New Exception("Read write not supported.")
            Else
                col = MyBase.SetUsageType(inputID, virtualInput, lineageID, usageType)
                AddOutputColumn(ComponentMetaData.OutputCollection(0).ID, col)
                AddOutputColumn(ComponentMetaData.OutputCollection(1).ID, col)
            End If
        End If

        Return col
    End Function

    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes")> _
    Public Overloads Overrides Sub ReinitializeMetaData()
        Dim distinctOutput As IDTSOutput100 = ComponentMetaData.OutputCollection(0)
        Dim duplicateOutput As IDTSOutput100 = ComponentMetaData.OutputCollection(1)
        If Not (distinctOutput.CustomPropertyCollection.Count = 0) Then
            Try
                Dim isDistinct As IDTSCustomProperty100 = distinctOutput.CustomPropertyCollection("IsDistinctOutput")
                If CType(isDistinct.Value, Boolean) = False Then
                    isDistinct.Value = True
                End If
            Catch
                AddIsDistinctCustomProperty(distinctOutput, True)
            End Try
        Else
            AddIsDistinctCustomProperty(distinctOutput, True)
        End If

        If Not (duplicateOutput.CustomPropertyCollection.Count = 0) Then
            Try
                Dim isDistinct As IDTSCustomProperty100 = duplicateOutput.CustomPropertyCollection("IsDistinctOutput")
                If CType(isDistinct.Value, Boolean) = True Then
                    isDistinct.Value = False
                End If
            Catch
                AddIsDistinctCustomProperty(duplicateOutput, False)
            End Try
        Else
            AddIsDistinctCustomProperty(duplicateOutput, False)
        End If

        AddInputColumnsToOutput(ComponentMetaData.OutputCollection(0).ID)
        AddInputColumnsToOutput(ComponentMetaData.OutputCollection(1).ID)
    End Sub

    Public Overloads Overrides Sub OnInputPathDetached(ByVal inputID As Integer)
        MyBase.OnInputPathDetached(inputID)
        ComponentMetaData.OutputCollection(0).OutputColumnCollection.RemoveAll()
        ComponentMetaData.OutputCollection(1).OutputColumnCollection.RemoveAll()
    End Sub

    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes")> _
    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId:="System.Exception.#ctor(System.String)")> _
    <CLSCompliant(False)> _
    Public Overloads Overrides Function InsertInput(ByVal insertPlacement As DTSInsertPlacement, ByVal inputID As Integer) As IDTSInput100
        Throw New Exception("Component " + ComponentMetaData.Name + " does not allow adding inputs.")
    End Function

    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes")> _
    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId:="System.Exception.#ctor(System.String)")> _
    <CLSCompliant(False)> _
    Public Overloads Overrides Function InsertOutput(ByVal insertPlacement As DTSInsertPlacement, ByVal outputID As Integer) As IDTSOutput100
        Throw New Exception("Component " + ComponentMetaData.Name + " does not allow adding outputs.")
    End Function

    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes")> _
    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId:="System.Exception.#ctor(System.String)")> _
    <CLSCompliant(False)> _
    Public Overloads Overrides Function InsertOutputColumnAt(ByVal outputID As Integer, ByVal outputColumnIndex As Integer, ByVal name As String, ByVal description As String) As IDTSOutputColumn100
        Throw New Exception("Output columns cannot be added to " + ComponentMetaData.Name)
    End Function

    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes")> _
    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId:="System.Exception.#ctor(System.String)")> _
    <CLSCompliant(False)> _
    Public Overloads Overrides Function SetOutputColumnProperty(ByVal outputID As Integer, ByVal outputColumnID As Integer, ByVal propertyName As String, ByVal propertyValue As Object) As IDTSCustomProperty100
        If propertyName = "InputColumnLineageID" Then
            Throw New Exception("The InputColumnLineageID property is read only.")
        End If

        Return MyBase.SetOutputColumnProperty(outputID, outputColumnID, propertyName, propertyValue)
    End Function

    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Usage", "CA2201:DoNotRaiseReservedExceptionTypes")> _
    <System.Diagnostics.CodeAnalysis.SuppressMessage("Microsoft.Globalization", "CA1303:DoNotPassLiteralsAsLocalizedParameters", MessageId:="System.Exception.#ctor(System.String)")> _
    <CLSCompliant(False)> _
    Public Overloads Overrides Function SetOutputProperty(ByVal outputID As Integer, ByVal propertyName As String, ByVal propertyValue As Object) As IDTSCustomProperty100
        If propertyName = "IsDistinctOutput" Then
            Throw New Exception("The IsDistinctOutput property is used internally by the component and cannot be modified.")
        End If

        Return MyBase.SetOutputProperty(outputID, propertyName, propertyValue)
    End Function

    Public Overloads Overrides Sub PreExecute()
        colInfoDistinctOutput = New ColumnInfos
        colInfoDistinctOutput.AddColumnInformation(BufferManager, ComponentMetaData.OutputCollection(0), ComponentMetaData.InputCollection(0))
        colInfoDuplicateOutput = New ColumnInfos
        colInfoDuplicateOutput.AddColumnInformation(BufferManager, ComponentMetaData.OutputCollection(1), ComponentMetaData.InputCollection(0))
        inputBufferCache = New Buffer
        inputBufferCache.ColumnInfos = colInfoDistinctOutput
    End Sub

    Public Overloads Overrides Sub ProcessInput(ByVal inputID As Integer, ByVal buffer As PipelineBuffer)
        If buffer Is Nothing Then
            Throw New ArgumentNullException("buffer")
        End If

        If Not buffer.EndOfRowset Then
            While buffer.NextRow
                inputBufferCache.AddRow(buffer)
            End While
        Else
            Dim startSort As DateTime = DateTime.Now
            ComponentMetaData.FireInformation(0, ComponentMetaData.Name, _
                "Starting sort operation, " + startSort.ToLongTimeString + ".", _
                "", 0, fireEventAgain)

            inputBufferCache.Sort()

            Dim sortComplete As DateTime = DateTime.Now
            Dim timeToSort As TimeSpan = sortComplete.Subtract(startSort)

            ComponentMetaData.FireInformation(0, ComponentMetaData.Name, _
                "Internal buffer cache sort complete," + sortComplete.ToLongTimeString + ", elapsed time was " + timeToSort.TotalSeconds.ToString(System.Globalization.CultureInfo.InvariantCulture) + " seconds.", _
                "", 0, fireEventAgain)

            Dim row As Integer = 0

            While row < inputBufferCache.RowCount
                If Not (row = 0) AndAlso inputBufferCache(row).Equals(inputBufferCache(row - 1)) Then
                    If Not (duplicatesBuffer Is Nothing) Then
                        AddRowToOutputBuffer(duplicatesBuffer, inputBufferCache(row), colInfoDuplicateOutput)
                    End If
                Else
                    If Not (distinctBuffer Is Nothing) Then
                        AddRowToOutputBuffer(distinctBuffer, inputBufferCache(row), colInfoDistinctOutput)
                    End If
                End If

                System.Math.Min(System.Threading.Interlocked.Increment(row), row - 1)
            End While

            If Not (distinctBuffer Is Nothing) Then
                distinctBuffer.SetEndOfRowset()
            End If

            If Not (duplicatesBuffer Is Nothing) Then
                duplicatesBuffer.SetEndOfRowset()
            End If
        End If
    End Sub

    Public Overloads Overrides Sub PrimeOutput(ByVal outputs As Integer, ByVal outputIDs As Integer(), ByVal buffers As PipelineBuffer())
        Dim x As Integer = 0
        While x < outputIDs.Length
            Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection.GetObjectByID(outputIDs(x))
            Dim isDistinctOutput As IDTSCustomProperty100 = output.CustomPropertyCollection("IsDistinctOutput")

            If CType(isDistinctOutput.Value, Boolean) Then
                distinctBuffer = buffers(x)
            Else
                duplicatesBuffer = buffers(x)
            End If

            System.Math.Min(System.Threading.Interlocked.Increment(x), x - 1)
        End While
    End Sub

    Private Sub InternalFireError(ByVal Message As String)
        ComponentMetaData.FireError(0, ComponentMetaData.Name, Message, "", 0, cancelEvent)
    End Sub

    Private Sub AddInputColumnsToOutput(ByVal outputID As Integer)
        Dim input As IDTSInput100 = ComponentMetaData.InputCollection(0)
        Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection.GetObjectByID(outputID)

        output.OutputColumnCollection.RemoveAll()
        For Each Col As IDTSInputColumn100 In input.InputColumnCollection
            AddOutputColumn(outputID, Col)
        Next
    End Sub

    Private Shared Sub AddRowToOutputBuffer(ByVal buffer As PipelineBuffer, ByVal row As Row, ByVal colInfos As ColumnInfos)
        buffer.AddRow()
        Dim col As Integer = 0

        While col < colInfos.Count
            buffer(colInfos(col).outputBufferColumnIndex) = row(col)
            System.Math.Min(System.Threading.Interlocked.Increment(col), col - 1)
        End While
    End Sub

    Private Sub AddOutputColumn(ByVal OutputID As Integer, ByVal inputColumn As IDTSInputColumn100)
        Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection.GetObjectByID(OutputID)
        Dim outputColumn As IDTSOutputColumn100 = output.OutputColumnCollection.New

        outputColumn.Name = inputColumn.Name

        Dim inputColLineageID As IDTSCustomProperty100 = outputColumn.CustomPropertyCollection.New

        inputColLineageID.Name = "InputColumnLineageID"
        inputColLineageID.Value = inputColumn.LineageID
        outputColumn.SetDataTypeProperties(inputColumn.DataType, inputColumn.Length, inputColumn.Precision, inputColumn.Scale, inputColumn.CodePage)
    End Sub

    Private Sub RemoveOutputColumn(ByVal OutputID As Integer, ByVal LineageID As Integer)
        Dim output As IDTSOutput100 = ComponentMetaData.OutputCollection.GetObjectByID(OutputID)
        For Each col As IDTSOutputColumn100 In output.OutputColumnCollection
            Dim inputColumnLineageID As IDTSCustomProperty100 = col.CustomPropertyCollection("InputColumnLineageID")

            If Not (inputColumnLineageID.Value Is Nothing) AndAlso CType(inputColumnLineageID.Value, Integer) = LineageID Then
                output.OutputColumnCollection.RemoveObjectByID(col.ID)
            End If
        Next
    End Sub

    Private Shared Function OutputColumnExists(ByVal output As IDTSOutput100, ByVal LineageID As Integer) As Boolean
        For Each outColumn As IDTSOutputColumn100 In output.OutputColumnCollection
            If Not (outColumn.CustomPropertyCollection.Count = 1) Then
                Return False
            End If

            If CType(outColumn.CustomPropertyCollection(0).Value, Integer) = LineageID Then
                Return True
            End If
        Next

        Return False
    End Function

    Private Shared Sub AddIsDistinctCustomProperty(ByVal output As IDTSOutput100, ByVal value As Object)
        Dim isDistinct As IDTSCustomProperty100 = output.CustomPropertyCollection.New
        isDistinct.Name = "IsDistinctOutput"
        isDistinct.Value = value
    End Sub
End Class